package belf.migrate.api.sink.jdbc;

import belf.migrate.api.sink.jdbc.helper.BeforeAfterComparor;
import belf.migrate.api.sink.jdbc.helper.ColumnRuleHelper;
import belf.migrate.api.sink.jdbc.helper.PrepareStatementCache;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.table.schema.CatalogTable;
import belf.migrate.api.table.schema.ColumnMetaData;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.taskconf.SinkConf;
import belf.migrate.api.type.LogLevelType;
import belf.migrate.api.util.PostLog;
import belf.migrate.core.date.DateTimeUtils;
import belf.migrate.core.date.DateUtils;
import belf.migrate.core.date.TimeUtils;
import belf.migrate.core.jdbc.DBFieldConvert;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.io.ParseException;
import org.locationtech.jts.io.WKBReader;

import java.io.ByteArrayInputStream;
import java.math.BigDecimal;
import java.sql.*;
import java.sql.Date;
import java.time.*;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.*;

import static java.sql.Types.*;

/**
 * 将从队列中获取的数据变动，批量插入到目标端数据库
 */
@Slf4j
public class JdbcBatchSaver {
    private final JobContext jobContext;

    private final AbstractSinkCatalog sinkCatalog;

    private final SinkConf sinkConf;

    private final Connection connection;

    private final ColumnRuleHelper columnRuleHelper;

    private final Gson gson;

    public JdbcBatchSaver(JobContext jobContext) {
        this.jobContext = jobContext;
        this.sinkCatalog = jobContext.getSinkCatalog();

        this.sinkConf = jobContext.getTaskConf().getSinkConf();
        connection = sinkCatalog.getConnection(sinkConf.getConnectionConf().getUrl());
        this.columnRuleHelper = new ColumnRuleHelper(jobContext.getTaskConf().getMapping());
        gson = new Gson();
/*        try {
            connection.setAutoCommit(true);
        } catch (SQLException e) {
            e.printStackTrace();
        }*/
    }

    public void save(List<JdbcSinkEvent> sinkEvents) throws SQLException {
        if (sinkEvents == null || sinkEvents.isEmpty()) return;

        JdbcSinkEvent firstEvent = sinkEvents.get(0);
        String ACTION = firstEvent.getEventType().getMessage();
        PreparedStatement preparedStatement = null;
        // 根据不同的EventType准备不同的SQL预处理语句
        switch (firstEvent.getEventType()) {
            case CREATE:
                preparedStatement = prepareInsertSQL(firstEvent);
                break;
            case UPDATE:
                preparedStatement = prepareUpdateSQL(firstEvent);
                break;
            case DELETE:
                preparedStatement = prepareDeleteSQL(firstEvent);
                break;
            case READ:
                // 对于"READ"操作，有两种处理办法：before为null，是插入；before非null，是更新
                if (firstEvent.getBefore() == null || firstEvent.getBefore().isEmpty()) {
                    preparedStatement = prepareInsertSQL(firstEvent);
                } else {
                    preparedStatement = prepareUpdateSQL(firstEvent);
                }

                break;
            default:
                break;
        }

        CatalogTable catalogTable = jobContext.getSinkTable(sinkConf.getCatalogName(),
                jobContext.getFullTableName(firstEvent.getTable()));
        //先关闭自增属性
        if (catalogTable.isAutoIncreament()) {
            sinkCatalog.autoIncreamentOff(catalogTable.getTablePath(), connection);
        }

        // 执行批量插入
        if (null != preparedStatement) {
            //ATTENTION：只有一个事件，必须使用executeUpdate；使用executeBatch会阻塞了
            if (sinkEvents.size() == 1) {
                setRowValue(preparedStatement, sinkEvents.get(0));
                preparedStatement.executeUpdate();
            } else if (sinkEvents.size() > 1) {
                for (JdbcSinkEvent event : sinkEvents) {
                    try {
                        setRowValue(preparedStatement, event);
                        preparedStatement.addBatch();
                    } catch (SQLException e) {
                        e.printStackTrace();
                        log.error("Prepare Batch {} Statement error: {}", ACTION, event);
                    }
                }
                preparedStatement.executeBatch();
                connection.commit();
            }

//            preparedStatement.close();
            log.debug("Batch {} {} Rows Successfully.", ACTION, sinkEvents.size());
            PostLog.sendLogModel(
                    Long.parseLong(jobContext.getJobId()),
                    "SYNC_DATA",
                    String.format("数据同步成功：%s行", sinkEvents.size()),
                    LogLevelType.INFO.getType(),
                    0,
                    jobContext.getProgress());
        }

        //事后打开自增属性
        if (catalogTable.isAutoIncreament()) {
            sinkCatalog.autoIncreamentOn(catalogTable.getTablePath(), connection);
        }
    }

    /**
     * INSERT ROW: 根据第一个事件准备 PreparedStatement预插入语句
     * @param event
     * @return
     * @throws SQLException
     */
    private PreparedStatement prepareInsertSQL(JdbcSinkEvent event) throws SQLException {
        PreparedStatement preparedStatement = PrepareStatementCache.getInstance()
                .get(jobContext.getJobId(), event.getDatabase(), event.getTable(), event.getEventType().getMessage());
        if (null != preparedStatement) {
            return preparedStatement;
        }

        StringBuilder sb = new StringBuilder();
        sb.append("INSERT INTO ");
        sb.append(jobContext.getFullTableName(event.getTable()));
        sb.append(" (");
        List<String> fieldNames = new ArrayList<>();
        event.getAfter().forEach(s -> fieldNames.add(s.getKey()));
        sb.append(String.join(",", fieldNames));
        sb.append(") VALUES(");
        List<String> placeholders = makeInsertPlaceholders(event);
        sb.append(String.join(",", placeholders));
        sb.append(")");
        String sql = sb.toString();
        log.debug("Prepared Insert SQL: {}", sql);

        preparedStatement = connection.prepareStatement(sql);
        PrepareStatementCache.getInstance().add(jobContext.getJobId(),
                event.getDatabase(),
                event.getTable(),
                event.getEventType().getMessage(),
                preparedStatement);
        return preparedStatement;
    }

    /**
     * UPDATE ROW: 根据第一个事件准备 PreparedStatement预更新语句
     * @param event 批处理的第一个事件
     * @return
     * @throws SQLException
     */
    private PreparedStatement prepareUpdateSQL(JdbcSinkEvent event) throws SQLException {

        List<KeyValue> difference = BeforeAfterComparor.subtract(event.getAfter(), event.getBefore());
        //判断是否有更新字段，如果没有，直接返回null
        if (difference.isEmpty()) {
            return null;
        }
        List<String> values = new ArrayList<>();
        difference.forEach(keyValue -> {
            values.add(keyValue.getKey() + "=?");
        });
        StringBuilder sb = new StringBuilder();
        sb.append("UPDATE ");
        sb.append(jobContext.getFullTableName(event.getTable()));
        sb.append(" SET ");
        String operation = event.getEventType().getMessage() + "-" +String.join(",", values);
        //update缓存prepareStatement需要判断更新的字段是否和之前一致
        PreparedStatement preparedStatement = PrepareStatementCache.getInstance()
                .get(jobContext.getJobId(), event.getDatabase(), event.getTable(), operation);
        if (null != preparedStatement) {
            return preparedStatement;
        }
        sb.append(String.join(",", values));
        sb.append(" WHERE ");
        boolean hasPrimaryKey = event.getKeys() != null && !(event.getKeys().isEmpty());
        if (hasPrimaryKey) {
            sb.append(toSQLCondition(event.getKeys()));
        } else {
            List<KeyValue> sameKeyValues = BeforeAfterComparor.intersection(event.getBefore(), event.getAfter());
            sb.append(toSQLCondition(sameKeyValues));
        }

        String sql = sb.toString();
        log.debug("Prepared Update SQL: {}", sql);

        preparedStatement = connection.prepareStatement(sql);
        PrepareStatementCache.getInstance().add(jobContext.getJobId(),
                event.getDatabase(),
                event.getTable(),
                event.getEventType().getMessage(),
                preparedStatement);
        return preparedStatement;
    }

    private PreparedStatement prepareDeleteSQL(JdbcSinkEvent event) throws SQLException {
        PreparedStatement preparedStatement = PrepareStatementCache.getInstance()
                .get(jobContext.getJobId(), event.getDatabase(), event.getTable(), event.getEventType().getMessage());
        if (null != preparedStatement) {
            return preparedStatement;
        }

        StringBuilder sb = new StringBuilder();
        sb.append("DELETE FROM ");
        sb.append(jobContext.getFullTableName(event.getTable()));
        sb.append(" WHERE ");
        if (event.getKeys() == null || event.getKeys().isEmpty()) { //无主键表的删除，用before做删除条件
            sb.append(toSQLCondition(event.getBefore()));
        } else { //有主键表的删除，用主键做条件
            sb.append(toSQLCondition(event.getKeys()));
        }

        String sql = sb.toString();
        log.debug("Prepared Delete SQL: {}", sql);

        preparedStatement = connection.prepareStatement(sql);
        PrepareStatementCache.getInstance().add(jobContext.getJobId(),
                event.getDatabase(),
                event.getTable(),
                event.getEventType().getMessage(),
                preparedStatement);
        return preparedStatement;
    }

    private List<String> makeInsertPlaceholders(JdbcSinkEvent event) {
        List<String> placeHolders = new ArrayList<>();
        event.getAfter().forEach(keyValue -> {
            placeHolders.add(columnRuleHelper.get_placeholder(event.getTable(), keyValue.getKey()));

        });
        return placeHolders;
    }

    private String toSQLCondition(List<KeyValue> keyValues) {
        List<String> results = new ArrayList<>();
        keyValues.forEach(keyValue -> {
            results.add(keyValue.getKey() + "=?");
/*            if (SQLTyperHelper.needQuote(keyValue.getValue())) {
                results.add(keyValue.getKey() + "=\"" + keyValue.getValue() + "\"");
            } else {
                results.add(keyValue.getKey() + "=" + keyValue.getValue());
            }*/

        });
        return String.join(" AND ", results);
    }

    /**
     * INSERT ROW: 用一行数据对PreparedStatement赋值
     * @param preparedStatement
     * @param event
     * @throws SQLException
     */
    private void setRowValue(PreparedStatement preparedStatement, JdbcSinkEvent event) throws SQLException {
        int parameterIndex = 1;
        switch (event.getEventType()) {
            case CREATE:
                for (KeyValue keyValue : event.getAfter()) {
                    setFieldValue(event, preparedStatement, parameterIndex++, keyValue.getKey(), keyValue.getValue());
                }
                break;
            case UPDATE:
                boolean hasPrimaryKey = event.getKeys() != null && !(event.getKeys().isEmpty());
                List<KeyValue> difference = BeforeAfterComparor.subtract(event.getAfter(), event.getBefore());
                List<KeyValue> conditionKeyValues;
                if (hasPrimaryKey) {
                    conditionKeyValues = event.getKeys();
                } else {
                    conditionKeyValues = BeforeAfterComparor.intersection(event.getBefore(), event.getAfter());
                }

                for (KeyValue keyValue : difference) {
                    setFieldValue(event, preparedStatement, parameterIndex++, keyValue.getKey(), keyValue.getValue());
                }
                for (KeyValue keyValue : conditionKeyValues) {
                    setFieldValue(event, preparedStatement, parameterIndex++, keyValue.getKey(), keyValue.getValue());
                }
                break;
            case DELETE:
                hasPrimaryKey = event.getKeys() != null && !(event.getKeys().isEmpty());
                if (hasPrimaryKey) {
                    conditionKeyValues = event.getKeys();
                } else {
                    conditionKeyValues = event.getBefore();
                }
                conditionKeyValues = new ArrayList<>();
                if (event.getKeys() == null || event.getKeys().isEmpty()) { //无主键表的删除，用before做删除条件
                    conditionKeyValues.addAll(event.getBefore());
                } else { //有主键表的删除，用主键做条件
                    conditionKeyValues.addAll(event.getKeys());
                }
                for (KeyValue keyValue : conditionKeyValues) {
                    setFieldValue(event, preparedStatement, parameterIndex++, keyValue.getKey(), keyValue.getValue());
                }
                break;
            case READ:
                List<KeyValue> allKeyValues = new ArrayList<>();
                if (event.getBefore() == null || event.getBefore().isEmpty()) { //无主键的插入事件
                    allKeyValues.addAll(event.getAfter());
                } else {  //无主键的更新事件
                    difference = BeforeAfterComparor.subtract(event.getAfter(), event.getBefore());
                    conditionKeyValues = BeforeAfterComparor.intersection(event.getBefore(), event.getAfter());
                    allKeyValues.addAll(difference);
                    allKeyValues.addAll(conditionKeyValues);
                }
                //无主键的插入事件
                for (KeyValue keyValue : allKeyValues) {
                    setFieldValue(event, preparedStatement, parameterIndex++, keyValue.getKey(), keyValue.getValue());
                }
                break;
            default:
                break;
        }
    }

    private void setFieldValue(JdbcSinkEvent event,
                               PreparedStatement preparedStatement,
                               int parameterIndex,
                               String columnName, Object value) throws SQLException {

        // Use reflection to get the corresponding setter method
        ColumnMetaData columnMetaData = getColumn(event, columnName);
        if (event.getTable().equalsIgnoreCase("address")) {
            log.info("columnMetaData: " + columnMetaData);
        }
        int jdbcType = columnMetaData.getColumnType();
        try {
            if (value == null) {
                preparedStatement.setNull(parameterIndex, jdbcType);
            } else if (columnRuleHelper.needConvert(event.getTable(), columnName)) {
                String conventValue = value.toString();
                Object convertedValue = columnRuleHelper.apply_read_rule(event.getTable(), columnName, conventValue);
                preparedStatement.setObject(parameterIndex, columnRuleHelper.apply_write_rule(event.getTable(), columnName, convertedValue));
            } else if (columnMetaData.getColumnClassName().equals("java.math.BigDecimal")) {
                preparedStatement.setBigDecimal(parameterIndex, BigDecimal.valueOf(Double.parseDouble(value.toString())));
            } else {
                switch (jdbcType) {
                    case BIT:
                        if (null == value) {
                            preparedStatement.setNull(parameterIndex, BIT);
                        }
                        preparedStatement.setInt(parameterIndex, ((int) value) & 0xFF);
                        break;
                    case BOOLEAN:
                        preparedStatement.setBoolean(parameterIndex, (boolean) value);
                        break;
                    case TINYINT:
                        if (value instanceof Integer) {
                            preparedStatement.setByte(parameterIndex, Byte.parseByte(value.toString()));
                        } else {
                            preparedStatement.setByte(parameterIndex, (byte) value);
                        }
                        break;
                    case SMALLINT:
                        if (value instanceof Integer) {
                            preparedStatement.setShort(parameterIndex, Short.parseShort(value.toString()));
                        } else {
                            preparedStatement.setShort(parameterIndex, (short) value);
                        }
                        break;
                    case INTEGER:
                        preparedStatement.setInt(parameterIndex, (int) value);
                        break;
                    case BIGINT:
                        preparedStatement.setLong(parameterIndex, (long) value);
                        break;
                    case FLOAT:
                        preparedStatement.setFloat(parameterIndex, (float) value);
                        break;
                    case REAL:
                    case DOUBLE:
                        preparedStatement.setDouble(parameterIndex, (double) value);
                        break;
                    case NUMERIC:
                        if (columnMetaData.getScale() == 0) {
                            if (value instanceof Integer) {
                                preparedStatement.setInt(parameterIndex, ((Integer) value).intValue());
                            } else if (value instanceof Long) {
                                preparedStatement.setLong(parameterIndex, ((Long) value).longValue());
                            } else if (value instanceof Double) {
                                preparedStatement.setDouble(parameterIndex, ((Double) value).doubleValue());
                            } else if (value instanceof Float) {
                                preparedStatement.setFloat(parameterIndex, ((Float) value).floatValue());
                            }
                        } else {
                            preparedStatement.setDouble(parameterIndex, (double) value);
                        }
                        break;
                    case DECIMAL:
                        preparedStatement.setBigDecimal(parameterIndex, BigDecimal.valueOf((Double) value));
                        break;
                    case CHAR:
                    case VARCHAR:
                    case LONGVARCHAR:
                    case NCHAR:
                    case NVARCHAR:
                    case LONGNVARCHAR:
                    case ROWID:
                        preparedStatement.setString(parameterIndex, (String) value);
                        break;
                    case BINARY:
                    case VARBINARY:
                    case LONGVARBINARY:
                        byte[] decodedBytes = Base64.getDecoder().decode((String) value);
                        preparedStatement.setBytes(parameterIndex, decodedBytes);
                        break;
                    case DATE:
                        if (columnMetaData.getColumnTypeName().equalsIgnoreCase("YEAR")
                                || value instanceof Integer) {
                            preparedStatement.setDate(parameterIndex, parseYear(value));
                        } else if (value instanceof Long lValue) {
                            preparedStatement.setDate(parameterIndex, new Date(lValue));
                        } else {
                            LocalDate localDate = DateUtils.parse((String) value);
                            preparedStatement.setDate(parameterIndex, Date.valueOf(localDate));
                        }
                        break;
                    case TIME:
                        LocalTime localTime = TimeUtils.parse((String) value, TimeUtils.Formatter.HH_MM_SS_SSS);
                        preparedStatement.setTime(parameterIndex, Time.valueOf(localTime));
                        break;
                    case TIMESTAMP:
                        String timeString = value.toString();
                        boolean isUTC;
                        //带'Z'的时间需要转成北京时间
                        try {
                            ZonedDateTime zonedDateTime = ZonedDateTime.parse(timeString, DateTimeFormatter.ISO_ZONED_DATE_TIME);
                            isUTC = zonedDateTime.getZone().equals(ZoneId.of("UTC")) || zonedDateTime.getZone().equals(ZoneId.of("Z"));
                            ZonedDateTime beijingTime = zonedDateTime.withZoneSameInstant(ZoneId.of("Asia/Shanghai"));
                            preparedStatement.setTimestamp(parameterIndex, java.sql.Timestamp.valueOf(beijingTime.toLocalDateTime()));
                        } catch (DateTimeParseException e1) {
                            isUTC = false;
                        }
                        if (!isUTC) {
                            //不是标准时间的Long类型的时间需要转成UTC标准时间
                            if (value instanceof Long) {
                                Instant instant = Instant.ofEpochMilli((Long) value);
                                ZonedDateTime utcTime = instant.atZone(ZoneId.of("UTC"));
                                preparedStatement.setTimestamp(parameterIndex, java.sql.Timestamp.valueOf(utcTime.toLocalDateTime()));
                            } else {
                                preparedStatement.setTimestamp(parameterIndex, parseTimestamp(value));
                            }
                        }
                        break;
                    case CLOB:
                    case NCLOB:
                        String largeText = (String) value;
                        preparedStatement.setCharacterStream(parameterIndex, new java.io.StringReader(largeText));
                        break;
                    case BLOB:
                        byte[] decodedBlob = Base64.getDecoder().decode((String) value);
                        ByteArrayInputStream bais = new ByteArrayInputStream(decodedBlob);
                        preparedStatement.setBlob(parameterIndex, bais);
                        break;
                    case TIME_WITH_TIMEZONE:
                        // 将时间戳转换为Instant对象
                        Instant instant = Instant.ofEpochMilli((long) value);
                        // 获取中国时区的ZoneOffset（UTC+08:00）
                        ZoneOffset chinaOffset = ZoneOffset.of("+08:00");

                        // 将Instant对象转换为OffsetTime对象，并设置为中国时区
                        OffsetTime timeInChina = OffsetTime.ofInstant(instant, chinaOffset);
                        preparedStatement.setObject(parameterIndex, timeInChina);
                        break;
                    case TIMESTAMP_WITH_TIMEZONE:
                        instant = Instant.ofEpochMilli((long) value);
                        chinaOffset = ZoneOffset.of("+08:00");
                        OffsetDateTime dateTimeInChina = instant.atOffset(chinaOffset);
                        preparedStatement.setObject(parameterIndex, dateTimeInChina);
                        break;
                    default:
                        preparedStatement.setObject(parameterIndex, value, JDBCType.valueOf(columnMetaData.getColumnType()));
                        break;
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            String message = String.format("Error setting value for parameter,table=%s,column=%s,value=%s, jdbcType=%s,columnTypeName=%s,parameterIndex=%d",
                    event.getTable(),
                    columnName,
                    value.toString(),
                    JDBCType.valueOf(columnMetaData.getColumnType()),
                    columnMetaData.getColumnTypeName(),
                    parameterIndex);
            log.error(message);
            throw new SQLException(message, e);
        }
    }

    private void setFieldValueByConvert(JdbcSinkEvent event,
                                        PreparedStatement preparedStatement,
                                        int parameterIndex,
                                        String columnName,
                                        Object value,
                                        DBFieldConvert.DBFieldConvertRule convertRule)
            throws SQLException {

        if (value != null || !(value instanceof ObjectNode)) {
            String inputValue = value.toString();
            // FIXME 暂时保证sqlserver-dm的geometry类型转换不报错，但是数据的坐标和源库不一致
            if (convertRule.getMethod().equalsIgnoreCase("base64_decode")) {
                byte[] outputValue = Base64.getDecoder().decode(inputValue);
                if (convertRule.getOutputType().equalsIgnoreCase("bytes")) {
                    try {
                        WKBReader reader = new WKBReader();
                        Geometry geometry = reader.read(outputValue);
                        preparedStatement.setString(parameterIndex, geometry.toText());
                    } catch (ParseException e) {
                        log.error("Error parsing geometry: " + e.getMessage(), e);
                    }
                    return;
                }
            }
        } else {
            ObjectNode valueNode = (ObjectNode) value;
            String inputValue = valueNode.get(convertRule.getJsonKey()).textValue();
            if (convertRule.getMethod().equalsIgnoreCase("base64_decode")) {
                byte[] outputValue = Base64.getDecoder().decode(inputValue);
                if (convertRule.getOutputType().equalsIgnoreCase("bytes")) {
                    preparedStatement.setBytes(parameterIndex, outputValue);
                    return;
                }
            }
        }

        preparedStatement.setNull(parameterIndex, NULL);
    }

    private java.sql.Timestamp parseTimestamp(Object value) {
        if (value instanceof Long) {
            return new java.sql.Timestamp((long)value);
        } else if (value instanceof Integer) {
            Integer intValue = (Integer) value;
            return new java.sql.Timestamp(intValue.longValue());// 转换为 Long 类型,在解析为Timestamp
        }

        //默认当作数字的字符串形式处理
        return Timestamp.valueOf(DateTimeUtils.parse(value.toString()));
    }

    private java.sql.Date parseYear(Object value) {
        if (value instanceof String && ((String) value).length() == 4) {
            String changedValue = value + "-01-01";
            return Date.valueOf(DateUtils.parse(changedValue));
        } else if (value instanceof Integer) {
            String changedValue = "" + value + "-01-01";
            return Date.valueOf(DateUtils.parse(changedValue));
        } else {
            return Date.valueOf(DateUtils.parse((String) value));
        }

    }
    private ColumnMetaData getColumn(JdbcSinkEvent event, String columnName) {
        CatalogTable catalogTable = jobContext.getSinkTable(sinkConf.getCatalogName(),
                jobContext.getFullTableName(event.getTable()));
        return catalogTable.getColumn(columnName);
    }

    // UPDATE WITH PRIMARY KEY

    // UPDATE WITHOUT PRIMARY KEY

    // DELETE
}
